
#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <pcap/pcap.h>
#include "ffvdp_recver.h"
#include "ffvdp_ring.h"

#include "av_frame_buf.h"
#include "av_pkt_buf.h"

#include <arpa/inet.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <libavcodec/avcodec.h>
#include "stream.h"

int ffvdp_packet_recver_recv(ffvdp_recver_t *recver, AVBufferRef **buf);
void ffvdp_packet_recver_destroy(ffvdp_recver_t *recver);

struct zerotime_frame_hdr_s {
    uint32_t recv_tv_sec;     /* seconds */
    uint32_t recv_tv_usec;    /* microseconds */
};

struct recver_priv_s {
    ffvdp_ring_t *frame_r;
    ffvdp_ring_t *r;
    char *url;
    const char *ifname;
    uint16_t ether_type;
    uint16_t snap_len;
    uint16_t timeout_ms;
    int is_promisc;
    char errbuf[PCAP_ERRBUF_SIZE];
    pcap_t *pcap;
    char pcap_filer[256];
    pthread_t recv_tid;
    pthread_t process_tid;

    av_pkt_buf_manager_t *pkt_buf_m;
    av_pkt_buf_t *pkt;
    av_frame_buf_manager_t *frame_buf_m;
    av_frame_buf_t *frame;

    av_pkt_buf_t *last_buf;
    uint16_t last_grp;
    uint32_t last_seq;
    int sof_found;
    int eof_found;

    double last_fetch_ts;
    double frame_time_ms;
    int frame_cache_nr;
    int fps;
};

// void _ffvdp_packet_process(struct recver_priv_s *p);
void *_ffvdp_packet_process(void *param);

struct priv {
    ffvdp_recver_t *recver;
    AVPacket *pkt;
};


static void *thread_recv(void *param) {
    struct recver_priv_s *p = param;

    int ret;
    struct pcap_pkthdr *pkt_header;
    const u_char *pkt_data;
    av_pkt_buf_t *buf;

    cpu_set_t cpuset;
    pthread_t thread;

    thread = pthread_self();

    /* Set affinity mask to include CPUs 0 to 7 */

    CPU_ZERO(&cpuset);
    CPU_SET(3, &cpuset);

    ret = pthread_setaffinity_np(thread, sizeof(cpu_set_t), &cpuset);

    while (1) {
        ret = pcap_next_ex(p->pcap, &pkt_header, &pkt_data);
        if (ret < 0) {
            pcap_perror(p->pcap, "Recv error");
            break;
        }

        if (ret == 0) {
            fprintf(stderr, "%s %s\n", __func__, "Recv timeout");
            //_ffvdp_packet_process(p);
            continue;
        }

        buf = av_pkt_buf_alloc(p->pkt_buf_m);
        if (!buf) {
            fprintf(stderr, "%s %s\n", __func__, "Out of memory, pkt drop");
            continue;
        } else {
            // fprintf(stderr, "%s %s %d\n", __func__, "recv pkt",
            // pkt_header->len);
            av_pkt_buf_add_data(buf, (char *)pkt_data, pkt_header->len);
            while (ffvdp_ring_full(p->r)) {
                fprintf(stderr, "%s\n", "ring full");
            }
#if 0
            fprintf(stderr, "ring count:%d\n", ffvdp_ring_count(p->r));
            int i;
            for(i =0; i < 14; i++) {
                fprintf(stderr, "%02x", buf->data[i]);
            }
            fprintf(stderr, "\n");
            struct packet_hdr_s *hdr;
            hdr = (struct packet_hdr_s*)&buf->data[0];
            fprintf(stderr, "s: %d, e: %d, grp: %d, seq %d\n", hdr->s, hdr->e, hdr->grp, ntohs(hdr->seq));
#endif
            ffvdp_ring_enqueue(p->r, buf);
        }
    }

    return NULL;
}

/*
 *   syntax of url:
 *      pcap:string:interger:interger:bool:interger
 *      pcap:ifname:ehter_type[:snap_len:promisc:read_timeout_ms]
 *      pacp:eth0:0x8888:65535:1:100
 *
 */
ffvdp_recver_t *ffvdp_packet_recver_create(const char *url) {
    fprintf(stderr, "%s %s\n", __func__, "entered");
    ffvdp_recver_t *recver;
    struct recver_priv_s *priv;
    recver = (ffvdp_recver_t *)calloc(
        1, sizeof(ffvdp_recver_t) + sizeof(struct recver_priv_s));
    if (!recver) {
        fprintf(stderr, "%s %s\n", __func__, "Out of memory");
        return NULL;
    }

    priv = (struct recver_priv_s *)(recver + 1);
    recver->priv = priv;
    priv->last_grp = -1;
    priv->last_seq = -1;

    priv->url = strdup(url);
    char *endprt;
    char *saveprt;
    char *token;
    token = strtok_r(priv->url, ":", &saveprt);
    if (!token) {
        fprintf(stderr, "%s %s\n", __func__, "Invalid parameter");
        goto create_error;
    } else if (strcmp(token, "pcap")) {
        fprintf(stderr, "%s %s\n", __func__, "Unsupported protocol");
        goto create_error;
    }

    priv->ifname = strtok_r(NULL, ":", &saveprt);
    if (!priv->ifname) {
        fprintf(stderr, "%s %s\n", __func__, "Invalid parameter");
        goto create_error;
    } else {
        fprintf(stdout, "%s interface %s\n", __func__, priv->ifname);
    }

    token = strtok_r(NULL, ":", &saveprt);
    if (!token) {
        fprintf(stderr, "%s %s\n", __func__, "Invalid parameter");
        goto create_error;
    } else {
        priv->ether_type = strtoul(token, &endprt, 0);
        fprintf(stdout, "%s ether_type 0x%x\n", __func__, priv->ether_type);
    }

    priv->snap_len = 65535;
    priv->is_promisc = 1;
    priv->timeout_ms = 0;
    token = strtok_r(NULL, ":", &saveprt);
    if (token) {
        priv->snap_len = strtoul(token, &endprt, 0);
        fprintf(stdout, "%s snap_len %d\n", __func__, priv->snap_len);
        token = strtok_r(NULL, ":", &saveprt);
        if (token) {
            priv->is_promisc = strtoul(token, &endprt, 0);
            fprintf(stdout, "%s is_promisc %d\n", __func__, !!priv->is_promisc);
            token = strtok_r(NULL, ":", &saveprt);
            if (token) {
                priv->timeout_ms = strtoul(token, &endprt, 0);
                fprintf(stdout, "%s timeout_ms %d\n", __func__,
                        priv->timeout_ms);
            }
        }
    }

    memset(priv->errbuf, 0, sizeof(priv->errbuf));
    // priv->pcap = pcap_open_live(priv->ifname, priv->snap_len,
    // priv->is_promisc,
    //                           priv->timeout_ms, priv->errbuf);
    priv->pcap = pcap_create(priv->ifname, priv->errbuf);
    if (!priv->pcap) {
        fprintf(stderr, "%s %s\n", __func__, priv->errbuf);
        goto create_error;
    } else if (strlen(priv->errbuf)) {
        fprintf(stderr, "%s %s\n", __func__, priv->errbuf);
    }

    pcap_set_snaplen(priv->pcap, priv->snap_len);
    pcap_set_promisc(priv->pcap, priv->is_promisc);
    pcap_set_timeout(priv->pcap, priv->timeout_ms);

    int ret;
    ret = pcap_set_buffer_size(priv->pcap, 5 * 1024 * 1024);
    if (ret != 0) {
        fprintf(stderr, "%s %s\n", __func__, "Failed to set pcap recv buffer");
    }

    ret = pcap_activate(priv->pcap);
    if (ret != 0) {
        fprintf(stderr, "%s %s\n", __func__, "Failed to activate pcap");
    }

    // pcap_setnonblock(priv->pcap, 1, priv->errbuf);
    pcap_setdirection(priv->pcap, PCAP_D_IN);

    snprintf(priv->pcap_filer, sizeof(priv->pcap_filer), "ether proto %d",
             priv->ether_type);
    struct bpf_program fp;
    ret = pcap_compile(priv->pcap, &fp, priv->pcap_filer, 1, 0);
    if (ret != 0) {
        fprintf(stderr, "%s %s\n", __func__, "Failed to compile pcap filter");
        goto create_error;
    }
    ret = pcap_setfilter(priv->pcap, &fp);
    if (ret != 0) {
        fprintf(stderr, "%s %s\n", __func__, "Failed to set pcap filter");
        goto create_error;
    }

    priv->r = ffvdp_ring_alloc(2048);
    if (!priv->r) {
        goto create_error;
    }

    priv->frame_r = ffvdp_ring_alloc(64);
    if (!priv->r) {
        goto create_error;
    }

    priv->pkt_buf_m = av_pkt_buf_alloc_manager(AV_PKT_BUF_TYPE_ZEROTIME_HI3516);
    assert(priv->pkt_buf_m);
    priv->frame_buf_m = av_frame_buf_alloc_manager(AV_FRAME_BUF_TYPE_GENERIC);
    assert(priv->frame_buf_m);

    pthread_create(&priv->recv_tid, NULL, thread_recv, priv);
    pthread_create(&priv->process_tid, NULL, _ffvdp_packet_process, priv);

    recver->recv = ffvdp_packet_recver_recv;
    recver->destroy = ffvdp_packet_recver_destroy;

    return recver;

create_error:
    ffvdp_packet_recver_destroy(recver);
    return NULL;
}

int ffvdp_packet_recver_recv(ffvdp_recver_t *recver, AVBufferRef **buf) {
    (void)recver;
    (void)buf;
#if 0
    struct recver_priv_s *priv;
    priv = (struct recver_priv_s *)recver->priv;
    while (ffvdp_ring_empty(priv->r)) {
        ;
    }

    ffvdp_ring_dequeue(priv->r, (void **)buf);
    assert(*buf);

    static FILE *fp = NULL;
    if (!fp) {
        fp = fopen("./test.265", "w");
    } 

    {
        fprintf(stderr, "%s %s %d\n", __func__, "buffer len", (*buf)->size);
        fwrite((*buf)->data, (*buf)->size, 1, fp);
        fflush(fp);
    }
#endif

    return -1;
}

void *_ffvdp_packet_process(void *param) {
    av_pkt_buf_t *buf;
    int sof;
    int eof;
    uint16_t grp;
    uint32_t seq;
    struct recver_priv_s *p;
    p = (struct recver_priv_s *)param;

    cpu_set_t cpuset;
    pthread_t thread;

    thread = pthread_self();

    /* Set affinity mask to include CPUs 0 to 7 */

    CPU_ZERO(&cpuset);
    CPU_SET(2, &cpuset);
    pthread_setaffinity_np(thread, sizeof(cpu_set_t), &cpuset);

    while (1) {
        if (p->last_buf) {
            buf = p->last_buf;
            p->last_buf = NULL;
        } else {
            if (ffvdp_ring_empty(p->r)) {
                continue;
            }

            ffvdp_ring_dequeue(p->r, (void **)&buf);
        }

        if (!p->frame) {
            p->frame = av_frame_buf_alloc(p->frame_buf_m);
            assert(p->frame);
        }

        grp = av_pkt_buf_grp(buf);
        seq = av_pkt_buf_seq(buf);
        sof = av_pkt_buf_is_sof(buf);
        eof = av_pkt_buf_is_eof(buf);

        // fprintf(stderr, "s: %d, e: %d, grp: %d, seq %d\n", sof, eof, grp,
        // seq);

        // fprintf(stderr, "ringsize: %d\n", ffvdp_ring_count(p->r));
        if (p->last_grp != grp && p->last_grp == (uint16_t)-1) {
            // first packet recved
            // grp num range [0..255], 256 is invalid
            p->last_grp = grp;
        }

        if (((p->last_seq + 1) & 0xffff) != seq) {
            if (p->last_seq == (uint32_t)-1) {
                p->last_seq = seq;
            }

            fprintf(stderr, "pkt lost, last_seq: %u, seq: %u\n", p->last_seq,
                    seq);
        }

        p->last_seq = seq;

        if (sof) {
            // fprintf(stderr, "Start of group %d found\n", hdr.grp);
            if (p->sof_found) {
                fprintf(stderr, "eof %d lost\n", p->last_grp);
                // TODO:
            }

            p->sof_found = 1;
            p->last_grp = grp;
            // TODO:  to process mutil group lost
        }

        if (!p->sof_found) {
            fprintf(stderr, "not sof_found, pkt dropped\n");
            av_pkt_buf_free(buf);
            continue;
        }

        if (p->last_grp != grp) {
            // the end of group of current group  lost,  recved packet of
            // diffrent group
            if (!sof) {
                fprintf(stderr, "no sof_found, pkt dropped\n");
                av_pkt_buf_free(buf);
                continue;
            }

            p->last_buf = buf;
            // fprintf(stderr, "last group %d,  new group %d\n", p->last_grp,
            // hdr.grp);
            continue;
        }

        av_frame_buf_add_data(p->frame, buf);
        p->last_grp = grp;
        av_pkt_buf_free(buf);

        if (eof) {
            // fprintf(stderr, "End of group %d found\n", hdr.grp);
            p->eof_found = 1;
            if (p->eof_found && p->sof_found) {
                double now_tv;
                double recv_tv;
                double trans_delay;
                struct zerotime_frame_hdr_s *fhdr;
                struct timeval ts;
                fhdr = (struct zerotime_frame_hdr_s*)(p->frame->data);
                gettimeofday(&ts, NULL);
                now_tv = ts.tv_sec * 1000 + ts.tv_usec / (double)1000;
                recv_tv = fhdr->recv_tv_sec * 1000 + fhdr->recv_tv_usec / (double)1000;
                trans_delay = (now_tv - recv_tv);
                //fprintf(stderr, "Frame %lu trans_delay: %f ms\n", p->frame->data_len, trans_delay);
                ffvdp_ring_enqueue(p->frame_r, p->frame);
                // fprintf(stderr, "Frame Size: %lu\n", p->frame->data_len);
                p->frame = NULL;
            } else {
                av_frame_buf_reset(p->frame);
            }
            p->sof_found = 0;
            p->eof_found = 0;
        }
    }
}

int ffvdp_packet_recver_recv_frame(ffvdp_recver_t *recver, AVPacket *pkt) {
    struct recver_priv_s *p;
    av_frame_buf_t *buf;
    uint32_t bufsize;

    p = (struct recver_priv_s *)recver->priv;

    if (ffvdp_ring_empty(p->frame_r)) {
        fprintf(stderr, "%s\n", "Waiting Frames ...");
        return -EAGAIN;
    } 

    struct zerotime_frame_hdr_s *fhdr;
    ffvdp_ring_dequeue(p->frame_r, (void **)&buf);

    fhdr = (struct zerotime_frame_hdr_s *)av_frame_buf_data(buf);
    bufsize = av_frame_buf_data_len(buf);

    memcpy(pkt->data, fhdr + 1, bufsize - sizeof(struct zerotime_frame_hdr_s));
    pkt->size = bufsize - sizeof(struct zerotime_frame_hdr_s);
    av_frame_buf_free(buf);
    return bufsize;
}

void ffvdp_packet_recver_destroy(ffvdp_recver_t *recver) {
    struct recver_priv_s *priv;
    priv = (struct recver_priv_s *)recver->priv;
    if (priv->pcap) {
        pcap_close(priv->pcap);
    }
    if (priv->r) {
        ffvdp_ring_free(priv->r);
    }
    free(recver);
}


static int fill_buffer(stream_t *s, char *buffer, int max_len)
{
    (void)max_len;

    struct priv *p = s->priv;
    int nread;
    do {
        nread = ffvdp_packet_recver_recv_frame(p->recver, p->pkt);
    } while(nread < 0);

    memcpy(buffer, p->pkt->data, p->pkt->size);

    return (int)p->pkt->size;
}

static int write_buffer(stream_t *s, char *buffer, int len)
{
    (void)s;
    (void)buffer;
    (void)len;
    return -1;
}

static int seek(stream_t *s, int64_t newpos)
{
    (void)s;
    (void)newpos;
    return 0;
}

static int control(stream_t *s, int cmd, void *arg)
{
    (void)s;
    (void)cmd;
    (void)arg;
    return STREAM_UNSUPPORTED;
}

static void s_close(stream_t *s)
{
    (void)s;
}


static int open_f(stream_t *stream)
{
    struct priv *p = talloc_ptrtype(stream, p);

    p->recver = ffvdp_packet_recver_create("pacp:eth0:0x8888:65535:1:100");
    if (!p->recver) {
        return STREAM_ERROR;
    }
    p->pkt = av_packet_alloc();
    av_new_packet(p->pkt, 500 * 1024);


    stream->seek = NULL;
    stream->seekable = false;

    stream->fast_skip = false;
    stream->fill_buffer = fill_buffer;
    stream->write_buffer = NULL;
    stream->control = control;
    stream->read_chunk = 64 * 1024;
    stream->close = s_close;
    stream->streaming = true;

    return STREAM_OK;
}

const stream_info_t stream_info_zerotime = {
    .name = "zerotime",
    .open = open_f,
    .protocols = (const char*const[]){ "zerotime", "", "hi3516", NULL },
    .can_write = false,
    .is_safe = true,
};
